<?php
/**
 * 阻塞队列 (Blocking Queue)
*  基于list的数据结构,使用 Redis 提供的 BRPOP 或 BLPOP 命令实现阻塞式消费。如果队列中没有消息，消费者会阻塞直到新消息到达。
 * 生产者：使用 LPUSH 向队列中添加消息。
 * 消费者：使用 RPOP 从队列中取出消息。
 * 添加数据 php mq/mqWaitList.php mq
 * 消费数据 php mq/mqWaitList.php
 */

class MqList
{
    /**
     * 添加数据
     */
    static function mq($data = ['task_id' => 2, 'message' => 'Blocking Queue Example'])
    {
        $redis = new Redis();
        $redis->connect('127.0.0.1', 6379);
        $data['task_id'] = date('YmdHis');
        $d = json_encode($data);
        $redis->lPush('queue:tasks', $d);
        echo "Message pushed to queue.\n";
    }

    /**
     * 消费数据
     */
    static function server()
    {
        $redis = new Redis();
        $redis->connect('127.0.0.1', 6379);
        // 设置 连接 永不超时时间
        // 不然 等待时间过长,redis服务端会关闭服务 Uncaught RedisException: read error on connection to 127.0.0.1:6379
        $redis->setOption(Redis::OPT_READ_TIMEOUT, -1);
        while (true) {
            //$data = [0 => key, value=> $v];
            //$keys 可以读取多个队列 $keys = ['queue:tasks']; $keys = '';
            $data = $redis->brPop('queue:tasks', 0); // 0 表示永远阻塞直到有消息
            if ($data) {
                $data = $data[1];
                $task = json_decode($data, true);
                //处理数据
                echo "消费任务 task: " . $task['task_id'] . "\n";
            }
        }
    }
}

if ('mq' ==  ($argv[1] ?? '') ||  'php-fpm' == PHP_SAPI) {
    MqList::mq();
    return;
}

MqList::server();







